package org.supermy.rocksdb;



        import java.io.IOException;
        import java.lang.Runnable;
        import java.lang.Math;
        import java.io.File;
        import java.lang.reflect.Constructor;
        import java.lang.reflect.InvocationTargetException;
        import java.nio.ByteBuffer;
        import java.nio.file.Files;
        import java.util.Collection;
        import java.util.Date;
        import java.util.EnumMap;
        import java.util.List;
        import java.util.Map;
        import java.util.Random;
        import java.util.concurrent.TimeUnit;
        import java.util.Arrays;
        import java.util.ArrayList;
        import java.util.concurrent.Callable;
        import java.util.concurrent.Executors;
        import java.util.concurrent.ExecutorService;
        import java.util.concurrent.Future;
        import java.util.concurrent.TimeUnit;
        import org.rocksdb.*;
        import org.rocksdb.RocksMemEnv;
        import org.rocksdb.util.SizeUnit;

class Stats {
    int id_;
    long start_;
    long finish_;
    double seconds_;
    long done_;
    long found_;
    long lastOpTime_;
    long nextReport_;
    long bytes_;
    StringBuilder message_;
    boolean excludeFromMerge_;

    // TODO(yhchiang): use the following arguments:
    //   (Long)Flag.stats_interval
    //   (Integer)Flag.stats_per_interval

    Stats(int id) {
        id_ = id;
        nextReport_ = 100;
        done_ = 0;
        bytes_ = 0;
        seconds_ = 0;
        start_ = System.nanoTime();
        lastOpTime_ = start_;
        finish_ = start_;
        found_ = 0;
        message_ = new StringBuilder("");
        excludeFromMerge_ = false;
    }

    void merge(final Stats other) {
        if (other.excludeFromMerge_) {
            return;
        }

        done_ += other.done_;
        found_ += other.found_;
        bytes_ += other.bytes_;
        seconds_ += other.seconds_;
        if (other.start_ < start_) start_ = other.start_;
        if (other.finish_ > finish_) finish_ = other.finish_;

        // Just keep the messages from one thread
        if (message_.length() == 0) {
            message_ = other.message_;
        }
    }

    void stop() {
        finish_ = System.nanoTime();
        seconds_ = (double) (finish_ - start_) * 1e-9;
    }

    void addMessage(String msg) {
        if (message_.length() > 0) {
            message_.append(" ");
        }
        message_.append(msg);
    }

    void setId(int id) { id_ = id; }
    void setExcludeFromMerge() { excludeFromMerge_ = true; }

    void finishedSingleOp(int bytes) {
        done_++;
        lastOpTime_ = System.nanoTime();
        bytes_ += bytes;
        if (done_ >= nextReport_) {
            if (nextReport_ < 1000) {
                nextReport_ += 100;
            } else if (nextReport_ < 5000) {
                nextReport_ += 500;
            } else if (nextReport_ < 10000) {
                nextReport_ += 1000;
            } else if (nextReport_ < 50000) {
                nextReport_ += 5000;
            } else if (nextReport_ < 100000) {
                nextReport_ += 10000;
            } else if (nextReport_ < 500000) {
                nextReport_ += 50000;
            } else {
                nextReport_ += 100000;
            }
            System.err.printf("... Task %s finished %d ops%30s\r", id_, done_, "");
        }
    }

    void report(String name) {
        // Pretend at least one op was done in case we are running a benchmark
        // that does not call FinishedSingleOp().
        if (done_ < 1) done_ = 1;

        StringBuilder extra = new StringBuilder("");
        if (bytes_ > 0) {
            // Rate is computed on actual elapsed time, not the sum of per-thread
            // elapsed times.
            double elapsed = (finish_ - start_) * 1e-9;
            extra.append(String.format("%6.1f MB/s", (bytes_ / 1048576.0) / elapsed));
        }
        extra.append(message_.toString());
        double elapsed = (finish_ - start_);
        double throughput = (double) done_ / (elapsed * 1e-9);

        System.out.format("%-12s : %11.3f micros/op %d ops/sec;%s%s\n",
                name, (elapsed * 1e-6) / done_,
                (long) throughput, (extra.length() == 0 ? "" : " "), extra.toString());
    }
}

public class DbBenchmark {
    enum Order {
        SEQUENTIAL,
        RANDOM
    }

    enum DBState {
        FRESH,
        EXISTING
    }

    static {
        RocksDB.loadLibrary();
    }

    abstract class BenchmarkTask implements Callable<Stats> {
        // TODO(yhchiang): use (Integer)Flag.perf_level.
        public BenchmarkTask(
                int tid, long randSeed, long numEntries, long keyRange) {
            tid_ = tid;
            rand_ = new Random(randSeed + tid * 1000);
            numEntries_ = numEntries;
            keyRange_ = keyRange;
            stats_ = new Stats(tid);
        }

        @Override public Stats call() throws RocksDBException {
            stats_.start_ = System.nanoTime();
            runTask();
            stats_.finish_ = System.nanoTime();
            return stats_;
        }

        abstract protected void runTask() throws RocksDBException;

        protected int tid_;
        protected Random rand_;
        protected long numEntries_;
        protected long keyRange_;
        protected Stats stats_;

        protected void getFixedKey(byte[] key, long sn) {
            generateKeyFromLong(key, sn);
        }

        protected void getRandomKey(byte[] key, long range) {
            generateKeyFromLong(key, Math.abs(rand_.nextLong() % range));
        }
    }

    abstract class WriteTask extends BenchmarkTask {
        public WriteTask(
                int tid, long randSeed, long numEntries, long keyRange,
                WriteOptions writeOpt, long entriesPerBatch) {
            super(tid, randSeed, numEntries, keyRange);
            writeOpt_ = writeOpt;
            entriesPerBatch_ = entriesPerBatch;
            maxWritesPerSecond_ = -1;
        }

        public WriteTask(
                int tid, long randSeed, long numEntries, long keyRange,
                WriteOptions writeOpt, long entriesPerBatch, long maxWritesPerSecond) {
            super(tid, randSeed, numEntries, keyRange);
            writeOpt_ = writeOpt;
            entriesPerBatch_ = entriesPerBatch;
            maxWritesPerSecond_ = maxWritesPerSecond;
        }

        @Override public void runTask() throws RocksDBException {
            if (numEntries_ != DbBenchmark.this.num_) {
                stats_.message_.append(String.format(" (%d ops)", numEntries_));
            }
            byte[] key = new byte[keySize_];
            byte[] value = new byte[valueSize_];

            try {
                if (entriesPerBatch_ == 1) {
                    for (long i = 0; i < numEntries_; ++i) {
                        getKey(key, i, keyRange_);
                        DbBenchmark.this.gen_.generate(value);
                        db_.put(writeOpt_, key, value);
                        stats_.finishedSingleOp(keySize_ + valueSize_);
                        writeRateControl(i);
                        if (isFinished()) {
                            return;
                        }
                    }
                } else {
                    for (long i = 0; i < numEntries_; i += entriesPerBatch_) {
                        WriteBatch batch = new WriteBatch();
                        for (long j = 0; j < entriesPerBatch_; j++) {
                            getKey(key, i + j, keyRange_);
                            DbBenchmark.this.gen_.generate(value);
                            batch.put(key, value);
                            stats_.finishedSingleOp(keySize_ + valueSize_);
                        }
                        db_.write(writeOpt_, batch);
                        batch.dispose();
                        writeRateControl(i);
                        if (isFinished()) {
                            return;
                        }
                    }
                }
            } catch (InterruptedException e) {
                // thread has been terminated.
            }
        }

        protected void writeRateControl(long writeCount)
                throws InterruptedException {
            if (maxWritesPerSecond_ <= 0) return;
            long minInterval =
                    writeCount * TimeUnit.SECONDS.toNanos(1) / maxWritesPerSecond_;
            long interval = System.nanoTime() - stats_.start_;
            if (minInterval - interval > TimeUnit.MILLISECONDS.toNanos(1)) {
                TimeUnit.NANOSECONDS.sleep(minInterval - interval);
            }
        }

        abstract protected void getKey(byte[] key, long id, long range);
        protected WriteOptions writeOpt_;
        protected long entriesPerBatch_;
        protected long maxWritesPerSecond_;
    }

    class WriteSequentialTask extends WriteTask {
        public WriteSequentialTask(
                int tid, long randSeed, long numEntries, long keyRange,
                WriteOptions writeOpt, long entriesPerBatch) {
            super(tid, randSeed, numEntries, keyRange,
                    writeOpt, entriesPerBatch);
        }
        public WriteSequentialTask(
                int tid, long randSeed, long numEntries, long keyRange,
                WriteOptions writeOpt, long entriesPerBatch,
                long maxWritesPerSecond) {
            super(tid, randSeed, numEntries, keyRange,
                    writeOpt, entriesPerBatch,
                    maxWritesPerSecond);
        }
        @Override protected void getKey(byte[] key, long id, long range) {
            getFixedKey(key, id);
        }
    }

    class WriteRandomTask extends WriteTask {
        public WriteRandomTask(
                int tid, long randSeed, long numEntries, long keyRange,
                WriteOptions writeOpt, long entriesPerBatch) {
            super(tid, randSeed, numEntries, keyRange,
                    writeOpt, entriesPerBatch);
        }
        public WriteRandomTask(
                int tid, long randSeed, long numEntries, long keyRange,
                WriteOptions writeOpt, long entriesPerBatch,
                long maxWritesPerSecond) {
            super(tid, randSeed, numEntries, keyRange,
                    writeOpt, entriesPerBatch,
                    maxWritesPerSecond);
        }
        @Override protected void getKey(byte[] key, long id, long range) {
            getRandomKey(key, range);
        }
    }

    class WriteUniqueRandomTask extends WriteTask {
        static final int MAX_BUFFER_SIZE = 10000000;
        public WriteUniqueRandomTask(
                int tid, long randSeed, long numEntries, long keyRange,
                WriteOptions writeOpt, long entriesPerBatch) {
            super(tid, randSeed, numEntries, keyRange,
                    writeOpt, entriesPerBatch);
            initRandomKeySequence();
        }
        public WriteUniqueRandomTask(
                int tid, long randSeed, long numEntries, long keyRange,
                WriteOptions writeOpt, long entriesPerBatch,
                long maxWritesPerSecond) {
            super(tid, randSeed, numEntries, keyRange,
                    writeOpt, entriesPerBatch,
                    maxWritesPerSecond);
            initRandomKeySequence();
        }
        @Override protected void getKey(byte[] key, long id, long range) {
            generateKeyFromLong(key, nextUniqueRandom());
        }

        protected void initRandomKeySequence() {
            bufferSize_ = MAX_BUFFER_SIZE;
            if (bufferSize_ > keyRange_) {
                bufferSize_ = (int) keyRange_;
            }
            currentKeyCount_ = bufferSize_;
            keyBuffer_ = new long[MAX_BUFFER_SIZE];
            for (int k = 0; k < bufferSize_; ++k) {
                keyBuffer_[k] = k;
            }
        }

        /**
         * Semi-randomly return the next unique key.  It is guaranteed to be
         * fully random if keyRange_ <= MAX_BUFFER_SIZE.
         */
        long nextUniqueRandom() {
            if (bufferSize_ == 0) {
                System.err.println("bufferSize_ == 0.");
                return 0;
            }
            int r = rand_.nextInt(bufferSize_);
            // randomly pick one from the keyBuffer
            long randKey = keyBuffer_[r];
            if (currentKeyCount_ < keyRange_) {
                // if we have not yet inserted all keys, insert next new key to [r].
                keyBuffer_[r] = currentKeyCount_++;
            } else {
                // move the last element to [r] and decrease the size by 1.
                keyBuffer_[r] = keyBuffer_[--bufferSize_];
            }
            return randKey;
        }

        int bufferSize_;
        long currentKeyCount_;
        long[] keyBuffer_;
    }

    class ReadRandomTask extends BenchmarkTask {
        public ReadRandomTask(
                int tid, long randSeed, long numEntries, long keyRange) {
            super(tid, randSeed, numEntries, keyRange);
        }
        @Override public void runTask() throws RocksDBException {
            byte[] key = new byte[keySize_];
            byte[] value = new byte[valueSize_];
            for (long i = 0; i < numEntries_; i++) {
                getRandomKey(key, keyRange_);
                int len = db_.get(key, value);
                if (len != RocksDB.NOT_FOUND) {
                    stats_.found_++;
                    stats_.finishedSingleOp(keySize_ + valueSize_);
                } else {
                    stats_.finishedSingleOp(keySize_);
                }
                if (isFinished()) {
                    return;
                }
            }
        }
    }

    class ReadSequentialTask extends BenchmarkTask {
        public ReadSequentialTask(
                int tid, long randSeed, long numEntries, long keyRange) {
            super(tid, randSeed, numEntries, keyRange);
        }
        @Override public void runTask() throws RocksDBException {
            RocksIterator iter = db_.newIterator();
            long i;
            for (iter.seekToFirst(), i = 0;
                 iter.isValid() && i < numEntries_;
                 iter.next(), ++i) {
                stats_.found_++;
                stats_.finishedSingleOp(iter.key().length + iter.value().length);
                if (isFinished()) {
                    iter.dispose();
                    return;
                }
            }
            iter.dispose();
        }
    }

    public DbBenchmark(Map<Flag, Object> flags) throws Exception {
        benchmarks_ = (List<String>) flags.get(Flag.benchmarks);
        num_ = (Integer) flags.get(Flag.num);
        threadNum_ = (Integer) flags.get(Flag.threads);
        reads_ = (Integer) (flags.get(Flag.reads) == null ?
                flags.get(Flag.num) : flags.get(Flag.reads));
        keySize_ = (Integer) flags.get(Flag.key_size);
        valueSize_ = (Integer) flags.get(Flag.value_size);
        compressionRatio_ = (Double) flags.get(Flag.compression_ratio);
        useExisting_ = (Boolean) flags.get(Flag.use_existing_db);
        randSeed_ = (Long) flags.get(Flag.seed);
        databaseDir_ = (String) flags.get(Flag.db);
        writesPerSeconds_ = (Integer) flags.get(Flag.writes_per_second);
        memtable_ = (String) flags.get(Flag.memtablerep);
        maxWriteBufferNumber_ = (Integer) flags.get(Flag.max_write_buffer_number);
        prefixSize_ = (Integer) flags.get(Flag.prefix_size);
        keysPerPrefix_ = (Integer) flags.get(Flag.keys_per_prefix);
        hashBucketCount_ = (Long) flags.get(Flag.hash_bucket_count);
        usePlainTable_ = (Boolean) flags.get(Flag.use_plain_table);
        useMemenv_ = (Boolean) flags.get(Flag.use_mem_env);
        flags_ = flags;
        finishLock_ = new Object();
        // options.setPrefixSize((Integer)flags_.get(Flag.prefix_size));
        // options.setKeysPerPrefix((Long)flags_.get(Flag.keys_per_prefix));
        compressionType_ = (String) flags.get(Flag.compression_type);
        compression_ = CompressionType.NO_COMPRESSION;
        try {
            if (compressionType_!=null) {
                final CompressionType compressionType =
                        CompressionType.getCompressionType(compressionType_);
                if (compressionType != null &&
                        compressionType != CompressionType.NO_COMPRESSION) {
                    System.loadLibrary(compressionType.getLibraryName());
                }

            }
        } catch (UnsatisfiedLinkError e) {
            System.err.format("Unable to load %s library:%s%n" +
                            "No compression is used.%n",
                    compressionType_, e.toString());
            compressionType_ = "none";
        }
        gen_ = new RandomGenerator(randSeed_, compressionRatio_);
    }

    private void prepareReadOptions(ReadOptions options) {
        options.setVerifyChecksums((Boolean)flags_.get(Flag.verify_checksum));
        options.setTailing((Boolean)flags_.get(Flag.use_tailing_iterator));
    }

    private void prepareWriteOptions(WriteOptions options) {
        options.setSync((Boolean)flags_.get(Flag.sync));
        options.setDisableWAL((Boolean)flags_.get(Flag.disable_wal));
    }

    private void prepareOptions(Options options) throws RocksDBException {
        if (!useExisting_) {
            options.setCreateIfMissing(true);
        } else {
            options.setCreateIfMissing(false);
        }
        if (useMemenv_) {
            options.setEnv(new RocksMemEnv());
        }
        switch (memtable_) {
            case "skip_list":
                options.setMemTableConfig(new SkipListMemTableConfig());
                break;
            case "vector":
                options.setMemTableConfig(new VectorMemTableConfig());
                break;
            case "hash_linkedlist":
                options.setMemTableConfig(
                        new HashLinkedListMemTableConfig()
                                .setBucketCount(hashBucketCount_));
                options.useFixedLengthPrefixExtractor(prefixSize_);
                break;
            case "hash_skiplist":
            case "prefix_hash":
                options.setMemTableConfig(
                        new HashSkipListMemTableConfig()
                                .setBucketCount(hashBucketCount_));
                options.useFixedLengthPrefixExtractor(prefixSize_);
                break;
            default:
                System.err.format(
                        "unable to detect the specified memtable, " +
                                "use the default memtable factory %s%n",
                        options.memTableFactoryName());
                break;
        }
        if (usePlainTable_) {
            options.setTableFormatConfig(
                    new PlainTableConfig().setKeySize(keySize_));
        } else {
            BlockBasedTableConfig table_options = new BlockBasedTableConfig();
            table_options.setBlockSize((Long)flags_.get(Flag.block_size))
                    .setBlockCacheSize((Long)flags_.get(Flag.cache_size))
                    .setCacheNumShardBits(
                            (Integer)flags_.get(Flag.cache_numshardbits));
            options.setTableFormatConfig(table_options);
        }
        options.setWriteBufferSize(
                (Long)flags_.get(Flag.write_buffer_size));
        options.setMaxWriteBufferNumber(
                (Integer)flags_.get(Flag.max_write_buffer_number));
        options.setMaxBackgroundCompactions(
                (Integer)flags_.get(Flag.max_background_compactions));
        options.getEnv().setBackgroundThreads(
                (Integer)flags_.get(Flag.max_background_compactions));
        options.setMaxBackgroundFlushes(
                (Integer)flags_.get(Flag.max_background_flushes));
        options.setMaxOpenFiles(
                (Integer)flags_.get(Flag.open_files));
        options.setUseFsync(
                (Boolean)flags_.get(Flag.use_fsync));
        options.setWalDir(
                (String)flags_.get(Flag.wal_dir));
        options.setDeleteObsoleteFilesPeriodMicros(
                (Integer)flags_.get(Flag.delete_obsolete_files_period_micros));
        options.setTableCacheNumshardbits(
                (Integer)flags_.get(Flag.table_cache_numshardbits));
        options.setAllowMmapReads(
                (Boolean)flags_.get(Flag.mmap_read));
        options.setAllowMmapWrites(
                (Boolean)flags_.get(Flag.mmap_write));
        options.setAdviseRandomOnOpen(
                (Boolean)flags_.get(Flag.advise_random_on_open));
        options.setUseAdaptiveMutex(
                (Boolean)flags_.get(Flag.use_adaptive_mutex));
        options.setBytesPerSync(
                (Long)flags_.get(Flag.bytes_per_sync));
        options.setBloomLocality(
                (Integer)flags_.get(Flag.bloom_locality));
        options.setMinWriteBufferNumberToMerge(
                (Integer)flags_.get(Flag.min_write_buffer_number_to_merge));
        options.setMemtablePrefixBloomSizeRatio((Double) flags_.get(Flag.memtable_bloom_size_ratio));
        options.setNumLevels(
                (Integer)flags_.get(Flag.num_levels));
        options.setTargetFileSizeBase(
                (Integer)flags_.get(Flag.target_file_size_base));
        options.setTargetFileSizeMultiplier((Integer)flags_.get(Flag.target_file_size_multiplier));
        options.setMaxBytesForLevelBase(
                (Integer)flags_.get(Flag.max_bytes_for_level_base));
        options.setMaxBytesForLevelMultiplier((Double) flags_.get(Flag.max_bytes_for_level_multiplier));
        options.setLevelZeroStopWritesTrigger(
                (Integer)flags_.get(Flag.level0_stop_writes_trigger));
        options.setLevelZeroSlowdownWritesTrigger(
                (Integer)flags_.get(Flag.level0_slowdown_writes_trigger));
        options.setLevelZeroFileNumCompactionTrigger(
                (Integer)flags_.get(Flag.level0_file_num_compaction_trigger));
        options.setMaxCompactionBytes(
                (Long) flags_.get(Flag.max_compaction_bytes));
        options.setDisableAutoCompactions(
                (Boolean)flags_.get(Flag.disable_auto_compactions));
        options.setMaxSuccessiveMerges(
                (Integer)flags_.get(Flag.max_successive_merges));
        options.setWalTtlSeconds((Long)flags_.get(Flag.wal_ttl_seconds));
        options.setWalSizeLimitMB((Long)flags_.get(Flag.wal_size_limit_MB));
        if(flags_.get(Flag.java_comparator) != null) {
            options.setComparator(
                    (AbstractComparator)flags_.get(Flag.java_comparator));
        }

    /* TODO(yhchiang): enable the following parameters
    options.setCompressionType((String)flags_.get(Flag.compression_type));
    options.setCompressionLevel((Integer)flags_.get(Flag.compression_level));
    options.setMinLevelToCompress((Integer)flags_.get(Flag.min_level_to_compress));
    options.setHdfs((String)flags_.get(Flag.hdfs)); // env
    options.setStatistics((Boolean)flags_.get(Flag.statistics));
    options.setUniversalSizeRatio(
        (Integer)flags_.get(Flag.universal_size_ratio));
    options.setUniversalMinMergeWidth(
        (Integer)flags_.get(Flag.universal_min_merge_width));
    options.setUniversalMaxMergeWidth(
        (Integer)flags_.get(Flag.universal_max_merge_width));
    options.setUniversalMaxSizeAmplificationPercent(
        (Integer)flags_.get(Flag.universal_max_size_amplification_percent));
    options.setUniversalCompressionSizePercent(
        (Integer)flags_.get(Flag.universal_compression_size_percent));
    // TODO(yhchiang): add RocksDB.openForReadOnly() to enable Flag.readonly
    // TODO(yhchiang): enable Flag.merge_operator by switch
    options.setAccessHintOnCompactionStart(
        (String)flags_.get(Flag.compaction_fadvice));
    // available values of fadvice are "NONE", "NORMAL", "SEQUENTIAL", "WILLNEED" for fadvice
    */
    }

    private void run() throws RocksDBException {
        if (!useExisting_) {
            destroyDb();
        }
        Options options = new Options();
        prepareOptions(options);
        open(options);

        printHeader(options);

        for (String benchmark : benchmarks_) {
            List<Callable<Stats>> tasks = new ArrayList<Callable<Stats>>();
            List<Callable<Stats>> bgTasks = new ArrayList<Callable<Stats>>();
            WriteOptions writeOpt = new WriteOptions();
            prepareWriteOptions(writeOpt);
            ReadOptions readOpt = new ReadOptions();
            prepareReadOptions(readOpt);
            int currentTaskId = 0;
            boolean known = true;

            switch (benchmark) {
                case "fillseq":
                    tasks.add(new WriteSequentialTask(
                            currentTaskId++, randSeed_, num_, num_, writeOpt, 1));
                    break;
                case "fillbatch":
                    tasks.add(new WriteRandomTask(
                            currentTaskId++, randSeed_, num_ / 1000, num_, writeOpt, 1000));
                    break;
                case "fillrandom":
                    tasks.add(new WriteRandomTask(
                            currentTaskId++, randSeed_, num_, num_, writeOpt, 1));
                    break;
                case "filluniquerandom":
                    tasks.add(new WriteUniqueRandomTask(
                            currentTaskId++, randSeed_, num_, num_, writeOpt, 1));
                    break;
                case "fillsync":
                    writeOpt.setSync(true);
                    tasks.add(new WriteRandomTask(
                            currentTaskId++, randSeed_, num_ / 1000, num_ / 1000,
                            writeOpt, 1));
                    break;
                case "readseq":
                    for (int t = 0; t < threadNum_; ++t) {
                        tasks.add(new ReadSequentialTask(
                                currentTaskId++, randSeed_, reads_ / threadNum_, num_));
                    }
                    break;
                case "readrandom":
                    for (int t = 0; t < threadNum_; ++t) {
                        tasks.add(new ReadRandomTask(
                                currentTaskId++, randSeed_, reads_ / threadNum_, num_));
                    }
                    break;
                case "readwhilewriting":
                    WriteTask writeTask = new WriteRandomTask(
                            -1, randSeed_, Long.MAX_VALUE, num_, writeOpt, 1, writesPerSeconds_);
                    writeTask.stats_.setExcludeFromMerge();
                    bgTasks.add(writeTask);
                    for (int t = 0; t < threadNum_; ++t) {
                        tasks.add(new ReadRandomTask(
                                currentTaskId++, randSeed_, reads_ / threadNum_, num_));
                    }
                    break;
                case "readhot":
                    for (int t = 0; t < threadNum_; ++t) {
                        tasks.add(new ReadRandomTask(
                                currentTaskId++, randSeed_, reads_ / threadNum_, num_ / 100));
                    }
                    break;
                case "delete":
                    destroyDb();
                    open(options);
                    break;
                default:
                    known = false;
                    System.err.println("Unknown benchmark: " + benchmark);
                    break;
            }
            if (known) {
                ExecutorService executor = Executors.newCachedThreadPool();
                ExecutorService bgExecutor = Executors.newCachedThreadPool();
                try {
                    // measure only the main executor time
                    List<Future<Stats>> bgResults = new ArrayList<Future<Stats>>();
                    for (Callable bgTask : bgTasks) {
                        bgResults.add(bgExecutor.submit(bgTask));
                    }
                    start();
                    List<Future<Stats>> results = executor.invokeAll(tasks);
                    executor.shutdown();
                    boolean finished = executor.awaitTermination(10, TimeUnit.SECONDS);
                    if (!finished) {
                        System.out.format(
                                "Benchmark %s was not finished before timeout.",
                                benchmark);
                        executor.shutdownNow();
                    }
                    setFinished(true);
                    bgExecutor.shutdown();
                    finished = bgExecutor.awaitTermination(10, TimeUnit.SECONDS);
                    if (!finished) {
                        System.out.format(
                                "Benchmark %s was not finished before timeout.",
                                benchmark);
                        bgExecutor.shutdownNow();
                    }

                    stop(benchmark, results, currentTaskId);
                } catch (InterruptedException e) {
                    System.err.println(e);
                }
            }
            writeOpt.dispose();
            readOpt.dispose();
        }
        options.dispose();
        db_.close();
    }

    private void printHeader(Options options) {
        int kKeySize = 16;
        System.out.printf("Keys:     %d bytes each\n", kKeySize);
        System.out.printf("Values:   %d bytes each (%d bytes after compression)\n",
                valueSize_,
                (int) (valueSize_ * compressionRatio_ + 0.5));
        System.out.printf("Entries:  %d\n", num_);
        System.out.printf("RawSize:  %.1f MB (estimated)\n",
                ((double)(kKeySize + valueSize_) * num_) / SizeUnit.MB);
        System.out.printf("FileSize:   %.1f MB (estimated)\n",
                (((kKeySize + valueSize_ * compressionRatio_) * num_) / SizeUnit.MB));
        System.out.format("Memtable Factory: %s%n", options.memTableFactoryName());
        System.out.format("Prefix:   %d bytes%n", prefixSize_);
        System.out.format("Compression: %s%n", compressionType_);
        printWarnings();
        System.out.printf("------------------------------------------------\n");
    }

    void printWarnings() {
        boolean assertsEnabled = false;
        assert assertsEnabled = true; // Intentional side effect!!!
        if (assertsEnabled) {
            System.out.printf(
                    "WARNING: Assertions are enabled; benchmarks unnecessarily slow\n");
        }
    }

    private void open(Options options) throws RocksDBException {
        System.out.println("Using database directory: " + databaseDir_);
        db_ = RocksDB.open(options, databaseDir_);
    }

    private void start() {
        setFinished(false);
        startTime_ = System.nanoTime();
    }

    private void stop(
            String benchmark, List<Future<Stats>> results, int concurrentThreads) {
        long endTime = System.nanoTime();
        double elapsedSeconds =
                1.0d * (endTime - startTime_) / TimeUnit.SECONDS.toNanos(1);

        Stats stats = new Stats(-1);
        int taskFinishedCount = 0;
        for (Future<Stats> result : results) {
            if (result.isDone()) {
                try {
                    Stats taskStats = result.get(3, TimeUnit.SECONDS);
                    if (!result.isCancelled()) {
                        taskFinishedCount++;
                    }
                    stats.merge(taskStats);
                } catch (Exception e) {
                    // then it's not successful, the output will indicate this
                }
            }
        }
        String extra = "";
        if (benchmark.indexOf("read") >= 0) {
            extra = String.format(" %d / %d found; ", stats.found_, stats.done_);
        } else {
            extra = String.format(" %d ops done; ", stats.done_);
        }

        System.out.printf(
                "%-16s : %11.5f micros/op; %6.1f MB/s;%s %d / %d task(s) finished.\n",
                benchmark, elapsedSeconds / stats.done_ * 1e6,
                (stats.bytes_ / 1048576.0) / elapsedSeconds, extra,
                taskFinishedCount, concurrentThreads);
    }

    public void generateKeyFromLong(byte[] slice, long n) {
        assert(n >= 0);
        int startPos = 0;

        if (keysPerPrefix_ > 0) {
            long numPrefix = (num_ + keysPerPrefix_ - 1) / keysPerPrefix_;
            long prefix = n % numPrefix;
            int bytesToFill = Math.min(prefixSize_, 8);
            for (int i = 0; i < bytesToFill; ++i) {
                slice[i] = (byte) (prefix % 256);
                prefix /= 256;
            }
            for (int i = 8; i < bytesToFill; ++i) {
                slice[i] = '0';
            }
            startPos = bytesToFill;
        }

        for (int i = slice.length - 1; i >= startPos; --i) {
            slice[i] = (byte) ('0' + (n % 10));
            n /= 10;
        }
    }

    private void destroyDb() {
        if (db_ != null) {
            db_.close();
        }
        // TODO(yhchiang): develop our own FileUtil
        // FileUtil.deleteDir(databaseDir_);
    }

    private void printStats() {
    }

    static void printHelp() {
        System.out.println("usage:");
        for (Flag flag : Flag.values()) {
            System.out.format("  --%s%n\t%s%n",
                    flag.name(),
                    flag.desc());
            if (flag.getDefaultValue() != null) {
                System.out.format("\tDEFAULT: %s%n",
                        flag.getDefaultValue().toString());
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Map<Flag, Object> flags = new EnumMap<Flag, Object>(Flag.class);
        for (Flag flag : Flag.values()) {
            if (flag.getDefaultValue() != null) {
                flags.put(flag, flag.getDefaultValue());
            }
        }
        for (String arg : args) {
            boolean valid = false;
            if (arg.equals("--help") || arg.equals("-h")) {
                printHelp();
                System.exit(0);
            }
            if (arg.startsWith("--")) {
                try {
                    String[] parts = arg.substring(2).split("=");
                    if (parts.length >= 1) {
                        Flag key = Flag.valueOf(parts[0]);
                        if (key != null) {
                            Object value = null;
                            if (parts.length >= 2) {
                                value = key.parseValue(parts[1]);
                            }
                            flags.put(key, value);
                            valid = true;
                        }
                    }
                }
                catch (Exception e) {
                }
            }
            if (!valid) {
                System.err.println("Invalid argument " + arg);
                System.exit(1);
            }
        }
        new DbBenchmark(flags).run();
    }

    private enum Flag {
        benchmarks(
                Arrays.asList(
                        "fillseq",
                        "readrandom",
                        "fillrandom"),
                "Comma-separated list of operations to run in the specified order\n" +
                        "\tActual benchmarks:\n" +
                        "\t\tfillseq          -- write N values in sequential key order in async mode.\n" +
                        "\t\tfillrandom       -- write N values in random key order in async mode.\n" +
                        "\t\tfillbatch        -- write N/1000 batch where each batch has 1000 values\n" +
                        "\t\t                   in random key order in sync mode.\n" +
                        "\t\tfillsync         -- write N/100 values in random key order in sync mode.\n" +
                        "\t\tfill100K         -- write N/1000 100K values in random order in async mode.\n" +
                        "\t\treadseq          -- read N times sequentially.\n" +
                        "\t\treadrandom       -- read N times in random order.\n" +
                        "\t\treadhot          -- read N times in random order from 1% section of DB.\n" +
                        "\t\treadwhilewriting -- measure the read performance of multiple readers\n" +
                        "\t\t                   with a bg single writer.  The write rate of the bg\n" +
                        "\t\t                   is capped by --writes_per_second.\n" +
                        "\tMeta Operations:\n" +
                        "\t\tdelete            -- delete DB") {
            @Override public Object parseValue(String value) {
                return new ArrayList<String>(Arrays.asList(value.split(",")));
            }
        },
        compression_ratio(0.5d,
                "Arrange to generate values that shrink to this fraction of\n" +
                        "\ttheir original size after compression.") {
            @Override public Object parseValue(String value) {
                return Double.parseDouble(value);
            }
        },
        use_existing_db(false,
                "If true, do not destroy the existing database.  If you set this\n" +
                        "\tflag and also specify a benchmark that wants a fresh database,\n" +
                        "\tthat benchmark will fail.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        num(1000000,
                "Number of key/values to place in database.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        threads(1,
                "Number of concurrent threads to run.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        reads(null,
                "Number of read operations to do.  If negative, do --nums reads.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        key_size(16,
                "The size of each key in bytes.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        value_size(100,
                "The size of each value in bytes.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        write_buffer_size(4L * SizeUnit.MB,
                "Number of bytes to buffer in memtable before compacting\n" +
                        "\t(initialized to default value by 'main'.)") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        max_write_buffer_number(2,
                "The number of in-memory memtables. Each memtable is of size\n" +
                        "\twrite_buffer_size.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        prefix_size(0, "Controls the prefix size for HashSkipList, HashLinkedList,\n" +
                "\tand plain table.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        keys_per_prefix(0, "Controls the average number of keys generated\n" +
                "\tper prefix, 0 means no special handling of the prefix,\n" +
                "\ti.e. use the prefix comes with the generated random number.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        memtablerep("skip_list",
                "The memtable format.  Available options are\n" +
                        "\tskip_list,\n" +
                        "\tvector,\n" +
                        "\thash_linkedlist,\n" +
                        "\thash_skiplist (prefix_hash.)") {
            @Override public Object parseValue(String value) {
                return value;
            }
        },
        hash_bucket_count(SizeUnit.MB,
                "The number of hash buckets used in the hash-bucket-based\n" +
                        "\tmemtables.  Memtables that currently support this argument are\n" +
                        "\thash_linkedlist and hash_skiplist.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        writes_per_second(10000,
                "The write-rate of the background writer used in the\n" +
                        "\t`readwhilewriting` benchmark.  Non-positive number indicates\n" +
                        "\tusing an unbounded write-rate in `readwhilewriting` benchmark.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        use_plain_table(false,
                "Use plain-table sst format.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        cache_size(-1L,
                "Number of bytes to use as a cache of uncompressed data.\n" +
                        "\tNegative means use default settings.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        seed(0L,
                "Seed base for random number generators.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        num_levels(7,
                "The total number of levels.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        numdistinct(1000L,
                "Number of distinct keys to use. Used in RandomWithVerify to\n" +
                        "\tread/write on fewer keys so that gets are more likely to find the\n" +
                        "\tkey and puts are more likely to update the same key.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        merge_keys(-1L,
                "Number of distinct keys to use for MergeRandom and\n" +
                        "\tReadRandomMergeRandom.\n" +
                        "\tIf negative, there will be FLAGS_num keys.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        bloom_locality(0,"Control bloom filter probes locality.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        duration(0,"Time in seconds for the random-ops tests to run.\n" +
                "\tWhen 0 then num & reads determine the test duration.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        num_multi_db(0,
                "Number of DBs used in the benchmark. 0 means single DB.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        histogram(false,"Print histogram of operation timings.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        min_write_buffer_number_to_merge(
                defaultOptions_.minWriteBufferNumberToMerge(),
                "The minimum number of write buffers that will be merged together\n" +
                        "\tbefore writing to storage. This is cheap because it is an\n" +
                        "\tin-memory merge. If this feature is not enabled, then all these\n" +
                        "\twrite buffers are flushed to L0 as separate files and this\n" +
                        "\tincreases read amplification because a get request has to check\n" +
                        "\tin all of these files. Also, an in-memory merge may result in\n" +
                        "\twriting less data to storage if there are duplicate records\n" +
                        "\tin each of these individual write buffers.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        max_background_compactions(
                defaultOptions_.maxBackgroundCompactions(),
                "The maximum number of concurrent background compactions\n" +
                        "\tthat can occur in parallel.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        max_background_flushes(
                defaultOptions_.maxBackgroundFlushes(),
                "The maximum number of concurrent background flushes\n" +
                        "\tthat can occur in parallel.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        /* TODO(yhchiang): enable the following
        compaction_style((int32_t) defaultOptions_.compactionStyle(),
            "style of compaction: level-based vs universal.") {
          @Override public Object parseValue(String value) {
            return Integer.parseInt(value);
          }
        },*/
        universal_size_ratio(0,
                "Percentage flexibility while comparing file size\n" +
                        "\t(for universal compaction only).") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        universal_min_merge_width(0,"The minimum number of files in a\n" +
                "\tsingle compaction run (for universal compaction only).") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        universal_max_merge_width(0,"The max number of files to compact\n" +
                "\tin universal style compaction.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        universal_max_size_amplification_percent(0,
                "The max size amplification for universal style compaction.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        universal_compression_size_percent(-1,
                "The percentage of the database to compress for universal\n" +
                        "\tcompaction. -1 means compress everything.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        block_size(defaultBlockBasedTableOptions_.blockSize(),
                "Number of bytes in a block.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        compressed_cache_size(-1L,
                "Number of bytes to use as a cache of compressed data.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        open_files(defaultOptions_.maxOpenFiles(),
                "Maximum number of files to keep open at the same time\n" +
                        "\t(use default if == 0)") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        bloom_bits(-1,"Bloom filter bits per key. Negative means\n" +
                "\tuse default settings.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        memtable_bloom_size_ratio(0.0d, "Ratio of memtable used by the bloom filter.\n"
                + "\t0 means no bloom filter.") {
            @Override public Object parseValue(String value) {
                return Double.parseDouble(value);
            }
        },
        cache_numshardbits(-1,"Number of shards for the block cache\n" +
                "\tis 2 ** cache_numshardbits. Negative means use default settings.\n" +
                "\tThis is applied only if FLAGS_cache_size is non-negative.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        verify_checksum(false,"Verify checksum for every block read\n" +
                "\tfrom storage.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        statistics(false,"Database statistics.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        writes(-1L, "Number of write operations to do. If negative, do\n" +
                "\t--num reads.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        sync(false,"Sync all writes to disk.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        use_fsync(false,"If true, issue fsync instead of fdatasync.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        disable_wal(false,"If true, do not write WAL for write.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        wal_dir("", "If not empty, use the given dir for WAL.") {
            @Override public Object parseValue(String value) {
                return value;
            }
        },
        target_file_size_base(2 * 1048576,"Target file size at level-1") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        target_file_size_multiplier(1,
                "A multiplier to compute target level-N file size (N >= 2)") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        max_bytes_for_level_base(10 * 1048576,
                "Max bytes for level-1") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        max_bytes_for_level_multiplier(10.0d,
                "A multiplier to compute max bytes for level-N (N >= 2)") {
            @Override public Object parseValue(String value) {
                return Double.parseDouble(value);
            }
        },
        level0_stop_writes_trigger(12,"Number of files in level-0\n" +
                "\tthat will trigger put stop.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        level0_slowdown_writes_trigger(8,"Number of files in level-0\n" +
                "\tthat will slow down writes.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        level0_file_num_compaction_trigger(4,"Number of files in level-0\n" +
                "\twhen compactions start.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        readwritepercent(90,"Ratio of reads to reads/writes (expressed\n" +
                "\tas percentage) for the ReadRandomWriteRandom workload. The\n" +
                "\tdefault value 90 means 90% operations out of all reads and writes\n" +
                "\toperations are reads. In other words, 9 gets for every 1 put.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        mergereadpercent(70,"Ratio of merges to merges&reads (expressed\n" +
                "\tas percentage) for the ReadRandomMergeRandom workload. The\n" +
                "\tdefault value 70 means 70% out of all read and merge operations\n" +
                "\tare merges. In other words, 7 merges for every 3 gets.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        deletepercent(2,"Percentage of deletes out of reads/writes/\n" +
                "\tdeletes (used in RandomWithVerify only). RandomWithVerify\n" +
                "\tcalculates writepercent as (100 - FLAGS_readwritepercent -\n" +
                "\tdeletepercent), so deletepercent must be smaller than (100 -\n" +
                "\tFLAGS_readwritepercent)") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        delete_obsolete_files_period_micros(0,"Option to delete\n" +
                "\tobsolete files periodically. 0 means that obsolete files are\n" +
                "\tdeleted after every compaction run.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        compression_type("snappy",
                "Algorithm used to compress the database.") {
            @Override public Object parseValue(String value) {
                return value;
            }
        },
        compression_level(-1,
                "Compression level. For zlib this should be -1 for the\n" +
                        "\tdefault level, or between 0 and 9.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        min_level_to_compress(-1,"If non-negative, compression starts\n" +
                "\tfrom this level. Levels with number < min_level_to_compress are\n" +
                "\tnot compressed. Otherwise, apply compression_type to\n" +
                "\tall levels.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        table_cache_numshardbits(4,"") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        stats_interval(0L, "Stats are reported every N operations when\n" +
                "\tthis is greater than zero. When 0 the interval grows over time.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        stats_per_interval(0,"Reports additional stats per interval when\n" +
                "\tthis is greater than 0.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        perf_level(0,"Level of perf collection.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        soft_rate_limit(0.0d,"") {
            @Override public Object parseValue(String value) {
                return Double.parseDouble(value);
            }
        },
        hard_rate_limit(0.0d,"When not equal to 0 this make threads\n" +
                "\tsleep at each stats reporting interval until the compaction\n" +
                "\tscore for all levels is less than or equal to this value.") {
            @Override public Object parseValue(String value) {
                return Double.parseDouble(value);
            }
        },
        rate_limit_delay_max_milliseconds(1000,
                "When hard_rate_limit is set then this is the max time a put will\n" +
                        "\tbe stalled.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        max_compaction_bytes(0L, "Limit number of bytes in one compaction to be lower than this\n" +
                "\threshold. But it's not guaranteed.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        readonly(false,"Run read only benchmarks.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        disable_auto_compactions(false,"Do not auto trigger compactions.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        wal_ttl_seconds(0L,"Set the TTL for the WAL Files in seconds.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        wal_size_limit_MB(0L,"Set the size limit for the WAL Files\n" +
                "\tin MB.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        /* TODO(yhchiang): enable the following
        direct_reads(rocksdb::EnvOptions().use_direct_reads,
            "Allow direct I/O reads.") {
          @Override public Object parseValue(String value) {
            return parseBoolean(value);
          }
          },
        direct_writes(rocksdb::EnvOptions().use_direct_reads,
          "Allow direct I/O reads.") {
          @Override public Object parseValue(String value) {
          return parseBoolean(value);
          }
          },
        */
        mmap_read(false,
                "Allow reads to occur via mmap-ing files.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        mmap_write(false,
                "Allow writes to occur via mmap-ing files.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        advise_random_on_open(defaultOptions_.adviseRandomOnOpen(),
                "Advise random access on table file open.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        compaction_fadvice("NORMAL",
                "Access pattern advice when a file is compacted.") {
            @Override public Object parseValue(String value) {
                return value;
            }
        },
        use_tailing_iterator(false,
                "Use tailing iterator to access a series of keys instead of get.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        use_adaptive_mutex(defaultOptions_.useAdaptiveMutex(),
                "Use adaptive mutex.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        bytes_per_sync(defaultOptions_.bytesPerSync(),
                "Allows OS to incrementally sync files to disk while they are\n" +
                        "\tbeing written, in the background. Issue one request for every\n" +
                        "\tbytes_per_sync written. 0 turns it off.") {
            @Override public Object parseValue(String value) {
                return Long.parseLong(value);
            }
        },
        filter_deletes(false," On true, deletes use bloom-filter and drop\n" +
                "\tthe delete if key not present.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        max_successive_merges(0,"Maximum number of successive merge\n" +
                "\toperations on a key in the memtable.") {
            @Override public Object parseValue(String value) {
                return Integer.parseInt(value);
            }
        },
        db(getTempDir("rocksdb-jni"),
                "Use the db with the following name.") {
            @Override public Object parseValue(String value) {
                return value;
            }
        },
        use_mem_env(false, "Use RocksMemEnv instead of default filesystem based\n" +
                "environment.") {
            @Override public Object parseValue(String value) {
                return parseBoolean(value);
            }
        },
        java_comparator(null, "Class name of a Java Comparator to use instead\n" +
                "\tof the default C++ ByteWiseComparatorImpl. Must be available on\n" +
                "\tthe classpath") {
            @Override
            protected Object parseValue(final String value) {
                try {
                    final ComparatorOptions copt = new ComparatorOptions();
                    final Class<AbstractComparator> clsComparator =
                            (Class<AbstractComparator>)Class.forName(value);
                    final Constructor cstr =
                            clsComparator.getConstructor(ComparatorOptions.class);
                    return cstr.newInstance(copt);
                } catch(final ClassNotFoundException cnfe) {
                    throw new IllegalArgumentException("Java Comparator '" + value + "'" +
                            " not found on the classpath", cnfe);
                } catch(final NoSuchMethodException nsme) {
                    throw new IllegalArgumentException("Java Comparator '" + value + "'" +
                            " does not have a public ComparatorOptions constructor", nsme);
                } catch(final IllegalAccessException | InstantiationException
                        | InvocationTargetException ie) {
                    throw new IllegalArgumentException("Unable to construct Java" +
                            " Comparator '" + value + "'", ie);
                }
            }
        };

        private Flag(Object defaultValue, String desc) {
            defaultValue_ = defaultValue;
            desc_ = desc;
        }

        public Object getDefaultValue() {
            return defaultValue_;
        }

        public String desc() {
            return desc_;
        }

        public boolean parseBoolean(String value) {
            if (value.equals("1")) {
                return true;
            } else if (value.equals("0")) {
                return false;
            }
            return Boolean.parseBoolean(value);
        }

        protected abstract Object parseValue(String value);

        private final Object defaultValue_;
        private final String desc_;
    }

    private final static String DEFAULT_TEMP_DIR = "/tmp";

    private static String getTempDir(final String dirName) {
        try {
            return Files.createTempDirectory(dirName).toAbsolutePath().toString();
        } catch(final IOException ioe) {
            System.err.println("Unable to create temp directory, defaulting to: " +
                    DEFAULT_TEMP_DIR);
            return DEFAULT_TEMP_DIR + File.pathSeparator + dirName;
        }
    }

    private static class RandomGenerator {
        private final byte[] data_;
        private int dataLength_;
        private int position_;
        private double compressionRatio_;
        Random rand_;

        private RandomGenerator(long seed, double compressionRatio) {
            // We use a limited amount of data over and over again and ensure
            // that it is larger than the compression window (32KB), and also
            byte[] value = new byte[100];
            // large enough to serve all typical value sizes we want to write.
            rand_ = new Random(seed);
            dataLength_ = value.length * 10000;
            data_ = new byte[dataLength_];
            compressionRatio_ = compressionRatio;
            int pos = 0;
            while (pos < dataLength_) {
                compressibleBytes(value);
                System.arraycopy(value, 0, data_, pos,
                        Math.min(value.length, dataLength_ - pos));
                pos += value.length;
            }
        }

        private void compressibleBytes(byte[] value) {
            int baseLength = value.length;
            if (compressionRatio_ < 1.0d) {
                baseLength = (int) (compressionRatio_ * value.length + 0.5);
            }
            if (baseLength <= 0) {
                baseLength = 1;
            }
            int pos;
            for (pos = 0; pos < baseLength; ++pos) {
                value[pos] = (byte) (' ' + rand_.nextInt(95));  // ' ' .. '~'
            }
            while (pos < value.length) {
                System.arraycopy(value, 0, value, pos,
                        Math.min(baseLength, value.length - pos));
                pos += baseLength;
            }
        }

        private void generate(byte[] value) {
            if (position_ + value.length > data_.length) {
                position_ = 0;
                assert(value.length <= data_.length);
            }
            position_ += value.length;
            System.arraycopy(data_, position_ - value.length,
                    value, 0, value.length);
        }
    }

    boolean isFinished() {
        synchronized(finishLock_) {
            return isFinished_;
        }
    }

    void setFinished(boolean flag) {
        synchronized(finishLock_) {
            isFinished_ = flag;
        }
    }

    RocksDB db_;
    final List<String> benchmarks_;
    final int num_;
    final int reads_;
    final int keySize_;
    final int valueSize_;
    final int threadNum_;
    final int writesPerSeconds_;
    final long randSeed_;
    final boolean useExisting_;
    final String databaseDir_;
    double compressionRatio_;
    RandomGenerator gen_;
    long startTime_;

    // env
    boolean useMemenv_;

    // memtable related
    final int maxWriteBufferNumber_;
    final int prefixSize_;
    final int keysPerPrefix_;
    final String memtable_;
    final long hashBucketCount_;

    // sst format related
    boolean usePlainTable_;

    Object finishLock_;
    boolean isFinished_;
    Map<Flag, Object> flags_;
    // as the scope of a static member equals to the scope of the problem,
    // we let its c++ pointer to be disposed in its finalizer.
    static Options defaultOptions_ = new Options();
    static BlockBasedTableConfig defaultBlockBasedTableOptions_ =
            new BlockBasedTableConfig();
    String compressionType_;
    CompressionType compression_;
}